// Copyright (c) Microsoft Corporation. All rights reserved. See License.txt in the project root for license information.
#include "pch.h"
#include "HttpRequestThrottler.h"

using namespace Weathr::Web;

using namespace concurrency;
using namespace std;
using namespace web::http;
using namespace web::http::client;

HttpRequestThrottler::HttpRequestThrottler(uint32_t maxRequestsPerSecond)
    : m_maxRequestsPerSecond(maxRequestsPerSecond)
    , m_activeRequests(0u)
    , m_requestHandler(new call<tuple<uri, task_completion_event<http_response>>>([this](tuple<uri, task_completion_event<http_response>> request) {
        HandleRequest(request);
}))
{
    m_waitEvent.set();
}

task<http_response> HttpRequestThrottler::DownloadAsync(const uri& requestUri)
{
    task_completion_event<http_response> tce;
    send(m_requests, make_tuple(requestUri, tce));
    return create_task(tce);
}

void HttpRequestThrottler::run()
{
    m_requests.link_target(m_requestHandler.get());
}

void HttpRequestThrottler::HandleRequest(const tuple<uri, task_completion_event<http_response>> &request)
{
    // Wait for a slot to be freed and for the time window to reopen.
    m_waitEvent.wait();

    if (++m_activeRequests == m_maxRequestsPerSecond)
    {
        // We've reached max capacity. Block any future requests until the current batch completes.
        m_waitEvent.reset();
    }

    // Perform the request.
    const auto uri = get<0>(request);
    const auto tce = get<1>(request);
    http_client client(uri);
    client.request(methods::GET).then([this, tce](task<http_response> previousTask)
    {
        try
        {
            // Set the completion event. This will complete the task returned by DownloadAsync.
            tce.set(previousTask.get());

            if (--m_activeRequests == 0u)
            {
                // The best we know is that the server processed the final request. To ensure that too many requests
                // are not submitted over 1 second, we must wait 1 second before releasing the waiter.
                // TODO: A better way to do this would be to query for when the server received the request. This is possible, 
                // but not through the current C++ REST SDK interface.
                concurrency::wait(1000u);
                m_waitEvent.set();
            }
        }
        catch (const http_exception& e)
        {
            // Possibly not connected.
            tce.set_exception(e);
            m_waitEvent.set();
        }
    });
}
